package com.cskaoyan.mall.order.mq;

import com.cskaoyan.mall.mq.constant.MqTopicConst;
import com.cskaoyan.mall.order.service.OrderService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

@Component
@Slf4j
public class DelayOrderConsumer {

    DefaultMQPushConsumer mqComsumer;

    @Value("${rocketmq.namesrv.addr}")
    String namesrvAddr;

    @Value("${rocketmq.consumer.group}")
    String consumerGroup;

    @Autowired
    OrderService orderService;

    @PostConstruct
    @SneakyThrows
    public void init(){
        // 初始化
        mqComsumer = new DefaultMQPushConsumer(consumerGroup + "-delayOrder");
        // 设置注册中心
        mqComsumer.setNamesrvAddr(namesrvAddr);
        // 订阅主题
        mqComsumer.subscribe(MqTopicConst.DELAY_ORDER_TOPIC, "*");
        // 设置消息监听器
        mqComsumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                String orderId = new String(messageExt.getBody());
                log.info("收到清单超时取消延迟消息，orderId:{}, messageId:{}",orderId,messageExt.getMsgId());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动
        mqComsumer.start();
        log.info("订单超时取消 消息消费者启动成功! ");
    }
}
